package io.github.eric.urpc.server;

import io.github.eric.urpc.core.conn.Connection;
import io.github.eric.urpc.core.exception.UrpcRuntimeException;
import io.github.eric.urpc.core.exception.UrpcMsgSendException;
import io.github.eric.urpc.core.rpc.Message;
import io.github.eric.urpc.core.rpc.RpcMappingResolver;
import io.github.eric.urpc.core.util.ConstantsUtil;
import io.github.eric.urpc.server.handler.ServerConnectionHandler;
import io.github.eric.urpc.server.handler.ServerMsgHandler;
import io.github.eric.urpc.server.handler.ServerStatHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.function.Predicate;
import java.util.stream.Collectors;

public class Server {
	private static final Logger LOGGER = LoggerFactory.getLogger(Server.class);

	private final ServerBootstrap serverBootstrap;
	private final EventLoopGroup bossGroup;
	private final EventLoopGroup workerGroup;
	private ChannelFuture serverChannelFuture;
	private final ServerContext serverContext;

	public Server(ServerConfig serverConfig) {
		this.serverContext = (ServerContext) serverConfig;
		this.serverContext.setLogger(LOGGER);
		if (serverContext.getMessageEncoder() == null) {
			throw new UrpcRuntimeException("[urpc server] message encoder is not present");
		}
		if (serverContext.getMessageDecoder() == null) {
			throw new UrpcRuntimeException("[urpc server] message decoder is not present");
		}
		if (serverContext.getRpcControllers() == null) {
			throw new UrpcRuntimeException("[urpc server] rpc controller is not present");
		}
		if (serverContext.getMessageRegistrar() == null) {
			throw new UrpcRuntimeException("[urpc server] message registrar is not present");
		}
		serverContext.setConnectionManager(new ServerConnectionManager(this.serverContext));
		new RpcMappingResolver().resolve(this.serverContext);

		bossGroup = new NioEventLoopGroup(new DefaultThreadFactory("urpc-server-boss-nioEventLoopGroup"));
		workerGroup = new NioEventLoopGroup(new DefaultThreadFactory("urpc-server-worker-nioEventLoopGroup"));
		serverBootstrap = new ServerBootstrap();

		ServerMsgHandler serverMsgHandler = new ServerMsgHandler(serverBootstrap, serverContext);
		ServerStatHandler serverStatHandler = new ServerStatHandler(serverBootstrap, serverContext);
		ServerConnectionHandler serverConnectionHandler = new ServerConnectionHandler(serverBootstrap, serverContext);
		LengthFieldPrepender lengthFieldPrepender = new LengthFieldPrepender(4);
		final LoggingHandler loggingHandler;
		if (serverContext.getNettyLoggingLevel() != null) {
			loggingHandler = new LoggingHandler(serverContext.getNettyLoggingLevel());
		} else {
			loggingHandler = null;
		}

		serverBootstrap.group(bossGroup, workerGroup)
				.channel(NioServerSocketChannel.class)
				.option(ChannelOption.SO_BACKLOG, serverContext.getSocketBacklog())
				.option(ChannelOption.SO_RCVBUF, serverContext.getSocketRecvBufSize())
				.childOption(ChannelOption.SO_RCVBUF, serverContext.getSocketRecvBufSize())
				.childOption(ChannelOption.SO_SNDBUF, serverContext.getSocketRecvBufSize())
				.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(64 * 1024, 256 * 1024))
				.childOption(ChannelOption.SO_KEEPALIVE, true)
				.childHandler(new ChannelInitializer<SocketChannel>() {
					@Override
					protected void initChannel(SocketChannel ch) throws Exception {
						if (loggingHandler != null) {
							ch.pipeline().addFirst(loggingHandler);
						}
						ch.pipeline().addLast(
								new IdleStateHandler(serverContext.getHeartBeatTimeout(), -1, -1),
								serverConnectionHandler,
								new LengthFieldBasedFrameDecoder(serverContext.getFrameMaxLength(),
										0, 4, 0,
										4),
								lengthFieldPrepender,
								// FOR TEST
//								new StringDecoder(),
//								new StringEncoder(),

								serverContext.getMessageEncoder(),
								serverContext.getMessageDecoder().copy(),
								serverMsgHandler,
								serverStatHandler
						);
					}
				});

		Thread shutdownHook = new Thread(new Thread(() -> {
			LOGGER.info("[urpc server] server shutdown hook started");
			shutdown();
			System.out.println("[urpc server] server shutdown hook ended");
			LOGGER.info("[urpc server] server shutdown hook ended");
		}),
				"urpc-server-shut-down-hook-thread");
		Runtime.getRuntime().addShutdownHook(shutdownHook);
	}

	public void start() {
		try {
			LOGGER.info("[urpc server] starting server ...");
			serverChannelFuture = this.serverBootstrap.bind(serverContext.getListenPort()).syncUninterruptibly();
			if (serverChannelFuture.channel().isActive()) {
				LOGGER.info("[urpc server] server started on port: " + serverContext.getListenPort() + " and listening ...");
			} else {
				throw serverChannelFuture.cause();
			}
		} catch (Throwable e) {
			if (serverChannelFuture != null) {
				serverChannelFuture.channel().close();
			}
		}
	}

	public void shutdown() {
		if (serverContext.isShuttingDown()) {
			return;
		}
		serverContext.shutdown();

		try {
			if (serverChannelFuture != null) {
				Future<?> future = serverChannelFuture.channel().close().syncUninterruptibly();
				if (future.isSuccess()) {
					LOGGER.info("[urpc server] server channels shut down successfully.");
				} else {
					LOGGER.warn("[urpc server] server channels shut down done, there may be exception.");
				}
			}
		} catch (Throwable t) {
			LOGGER.error("[urpc server] server channels shut down error", t);
		}

		try {
			if (bossGroup != null) {
				Future<?> future = bossGroup.shutdownGracefully().syncUninterruptibly();
				if (future.isSuccess()) {
					LOGGER.info("[urpc server] server boss group shut down successfully.");
				} else {
					LOGGER.warn("[urpc server] server boss group shut down done, there may be exception.");
				}
			}
		} catch (Throwable t) {
			LOGGER.error("[urpc server] server boss group shut down error", t);
		}

		try {
			if (workerGroup != null) {
				Future<?> future = workerGroup.shutdownGracefully().syncUninterruptibly();
				if (future.isSuccess()) {
					LOGGER.info("[urpc server] server worker group shut down successfully.");
				} else {
					LOGGER.warn("[urpc server] server worker group shut down done, there may be exception.");
				}
			}
		} catch (Throwable t) {
			LOGGER.error("[urpc server] server worker group shut down error", t);
		}

		try {
			if (serverContext.getHashedWheelTimer() != null) {
				serverContext.getHashedWheelTimer().stop();
				LOGGER.info("[urpc server] server hashed wheel timer stop.");
			}
		} catch (Exception e) {
			LOGGER.info("[urpc server] server hashed wheel timer stop error.", e);
		}

		try {
			serverContext.getRpcExecutor().shutdown();
		} catch (Exception e) {
			LOGGER.info("[urpc server] server rpc executor stop error.", e);
		}
	}

	public ServerConfig getServerConfig() {
		return serverContext;
	}

	/**
	 * 向符合条件的第一个连接发送异步消息
	 */
	public void sendAsync(Message message, Predicate<Connection> predicate) {
		Collection<Connection> connections = serverContext.getConnectionManager().getConnections(predicate);
		if (connections.isEmpty()) {
			throw new UrpcMsgSendException("[urpc server] server can not get any available connection");
		}
		Connection conn = connections.iterator().next();
		conn.sendAsync(message);
	}

	/**
	 * 向applicationName中的某一个连接发送异步消息（多次发送可能落在不同实例上）
	 */
	public void sendAsyncToOneInstanceByApplicationName(Message message, String applicationName) {
		Collection<Connection> connections = serverContext.getConnectionManager().getConnectionsByApplicationName(applicationName);
		if (connections.isEmpty()) {
			throw new UrpcMsgSendException(String.format("[urpc server] server can not get any available connection by application name %s", applicationName));
		}
		Connection conn = connections.iterator().next();
		conn.sendAsync(message);
	}

	/**
	 * 向applicationName中的所有实例发送异步消息（每个实例只发一个消息）
	 */
	public void sendAsyncToAllInstancesByApplicationName(Message message, String applicationName) {
		Collection<Connection> connections = serverContext.getConnectionManager().getConnectionsByApplicationName(applicationName);
		if (connections.isEmpty()) {
			throw new UrpcMsgSendException(String.format("[urpc server] server can not get any available connection by application name %s", applicationName));
		}
		Set<String> alreadySentApplicationId = new HashSet<>();
		for (Connection conn : connections) {
			String applicationId = (String) conn.getAttribute(ConstantsUtil.APPLICATION_ID);
			if (alreadySentApplicationId.contains(applicationId))	{
				continue;
			}
			try {
				conn.sendAsync(message);
				alreadySentApplicationId.add(applicationName);
			} catch (Exception e) {
				LOGGER.warn("[urpc server] server send async message {} to {} by application name error, ignore it and go on",
						message.getClass().getSimpleName(), conn.getRemoteAddress().getHostString());
			}
		}
	}

	/**
	 * 向applicationName中的符合条件的连接发送消息
	 */
	public void sendAsyncByApplicationName(Message message, String applicationName, Predicate<Connection> predicate) {
		Collection<Connection> connections = serverContext.getConnectionManager().getConnectionsByApplicationName(applicationName);
		if (predicate != null) {
			connections = connections.stream().filter(predicate).collect(Collectors.toList());
		}
		if (connections.isEmpty()) {
			throw new UrpcMsgSendException(String.format("[urpc server] server can not get any available connection by application name %s", applicationName));
		}
		for (Connection conn : connections) {
			try {
				conn.sendAsync(message);
			} catch (Exception e) {
				LOGGER.warn("[urpc server] server send async message {} to {} by application name and predicate error, ignore it and go on",
						message.getClass().getSimpleName(), conn.getRemoteAddress().getHostString());
			}
		}
	}

	/**
	 * 向applicationName中的指定实例发送异步消息（多次发送始终落在同一个实例上）
	 */
	public void sendAsyncToOneInstanceByApplicationNameAndId(Message message, String applicationName, String applicationId) {
		Collection<Connection> connections = serverContext.getConnectionManager().getConnectionsByApplicationNameAndId(applicationName, applicationId);
		if (connections == null) {
			throw new UrpcMsgSendException(String.format("[urpc server] server can not get any available connection by application name %s and id %s",
					applicationName, applicationId));
		}
		Connection conn = connections.iterator().next();
		conn.sendAsync(message);
	}

	public void broadcastAsync(Message message) {
		broadcastAsync(message, null);
	}

	public void broadcastAsync(Message message, Predicate<Connection> predicate) {
		Collection<Connection> connections;
		if (predicate == null) {
			connections = serverContext.getConnectionManager().getConnections();
		} else {
			connections = serverContext.getConnectionManager().getConnections(predicate);
		}
		for (Connection conn : connections) {
			try {
				conn.sendAsync(message);
			} catch (Exception e) {
				LOGGER.warn("[urpc server] server broadcast async message {} to {} error, ignore it and go on",
						message.getClass().getSimpleName(), conn.getRemoteAddress().getHostString());
			}
		}
	}

	public void setConnectionAttributes(String connId, Map<String, Object> attributes) {
		serverContext.getConnectionManager().setConnectionAttributes(connId, attributes);
	}
}
